/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2013_05_21			*/
/*	File:		MrHub.cpp			*/
/********************************************************/

#include "MrHub.h"

CMrThash g_CThash;
bool g_bDebug = false;

int main(int argc, char* argv[])
{
	CMrTcp CTcp;
	int iIndex = 0;
	int iReturn = 0;
	CMrThread CAdmin;
	char szPort[MR_VALUE] = { 0 };
	char szConf[MR_VALUE] = "MrRed.conf";

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:dhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':
				strncpy(szConf, optarg, MR_VALUE);
				break;
			case 'd':
				g_bDebug = true;
				break;
			case 'h':
				MrHubHelp(argv[0]);
				return MR_SUCCESS;
			case 'v':
				MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
				MR_PRINT("%s Build on %s %s.\n", MR_HB_VERSION, __DATE__, __TIME__);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	signal(SIGSEGV, MrTkQuit);
	signal(SIGTERM, MrTkQuit);

	/* config */
	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrTkConfig(szConf, MR_HB_CONFIG, szPort)) {
		if (true == g_bDebug) {
			MR_DEBUG("Port: %s\n", szPort);
		}
	}
	else {
		return MR_FAILURE;
	}

	/* fork */
	if (true != g_bDebug) {
		MrFork();
	}

	/* admin */
	if (true == g_bDebug) {
		MR_DEBUG("Create admin thread\n");
	}
	CAdmin.MrThreadSetStartRoutine(MrHubAdmin);
	if (MR_SUCCESS != CAdmin.MrThreadCreate(NULL)) {
		return MR_FAILURE;
	}

	/* thash */
	if (MR_SUCCESS == g_CThash.MrThashCreate(g_CThash.MrIpcKey(MR_HB_PATH, atoi(szPort) % MR_HB_KEY), MR_HB_THASH)) {
		/* accept */
		if (true == g_bDebug) {
			MR_DEBUG("MrHub will listen...\n");
		}
		CTcp.MrNetLimit(MR_SOCK_MAX);
		if (MR_SUCCESS == CTcp.MrTcpInit(szPort)) {
			CTcp.MrTcpSetFunction(MrHubRun);
			CTcp.MrTcpAccept();
			CTcp.MrNetClose();
		}
		//g_CThash.MrThashDelete();
	}
	else {
		MR_ERROR("Failed to MrThashCreate\n");
	}

	return MR_SUCCESS;
}

void MrHubHelp(char* pszProc)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[*] -c or --conf:       set configure.\n");
	MR_PRINT("[0] -d or --debug:      set debug mode.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");
	MR_PRINT("[0] -v or --version:    show version.\n");
	MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
	MR_PRINT("%s Build on %s %s.\n", MR_HB_VERSION, __DATE__, __TIME__);

	return;
}

void MrHubRun(int iSocket)
{
	CMrThread CThread;

	CThread.MrThreadAttrInit();
	CThread.MrThreadAttrSetDetach();
	CThread.MrThreadSetStartRoutine(MrHubAccept);
	CThread.MrThreadCreate((void*)iSocket);

	return;
}

void* MrHubAdmin(void* pNULL)
{
	CMrSet CSet;
	ULONG ulIndex = 0;
	ULONG ulCount = 0;
	ULONG ulTotal = 0;
	time_t ulTime = 0;
	MRHUBDATA_S* pstData = NULL;
	CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>::MR_ITE iteCur;
	CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>** ppCTmap = NULL;

	while (true) {
		sleep(MR_HB_SLEEP);
		for (ulCount = 0; ulCount < MR_HB_THASH; ulCount ++) {
			ppCTmap = (CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>**)g_CThash.MrHashGetBucket(ulCount);
			if (NULL != *ppCTmap && (ulTotal = (*ppCTmap)->MrMapSize()) > 0) {
				if (MR_SUCCESS == CSet.MrSetInit(ulTotal, sizeof(ULONGLONG))) {
					/* check */
					ulIndex = 0;
					ulTime = time(NULL);
					(*ppCTmap)->MrTmapReadLock();
					for (iteCur = (*ppCTmap)->MrMapBegin(); iteCur != (*ppCTmap)->MrMapEnd(); iteCur ++) {
						pstData = (MRHUBDATA_S*)&iteCur->second;
						if (ulTime - pstData->m_ulStamp > MR_HB_SLEEP) {
							CSet.MrSetIndexPut(ulIndex ++, (void*)&iteCur->first);
							MrFree((void**)&pstData->m_puszData);
						}
					}
					(*ppCTmap)->MrTmapUnlock();

					/* clear */
					for (ulIndex = 0; ulIndex < CSet.MrSetGetCount(); ulIndex ++) {
						(*ppCTmap)->MrTmapDelete(*(ULONGLONG*)CSet.MrSetIndexGet(ulIndex));
					}

					CSet.MrSetDestroy();
				}
				else {
					MR_ERROR("%s\n", strerror(errno));
				}
			}
		}
	}

	pthread_exit(NULL);
}

void* MrHubAccept(void* pSocket)
{
	CMrPro CPro;
	CMrTcp CTcp;
	CMrTime CTime;
	USHORT usTotal = 0;
	MRHUBDATA_S stData;
	ULONGLONG ullKey = 0;
	MRPROKV_S stKv[MR_MR_VAL];
	char szKey[MR_VALUE] = { 0 };
	UCHAR ucResult = MR_DEF_SUCC;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };
	CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>** ppCTmap = NULL;

	signal(SIGPIPE, SIG_IGN);
	CTcp.MrNetSetSocket((long)pSocket);
	memset(&stData, '\0', sizeof(MRHUBDATA_S));

	if (MR_PRO_HEAD == CTcp.MrNetRecv(uszHead, MR_PRO_HEAD)) {
		if (true == g_bDebug) {
			MR_PRINT("\n");
			CTime.MrTimeOn();
			MR_DEBUG("Thread[%lu] start\n", pthread_self());
		}

		/* head */
		CPro.MrProParse(uszHead, MR_PRO_HEAD);
		if (true == g_bDebug) {
			CPro.MrProDisplay();
		}

		/* body */
		if ((stData.m_iLength = CPro.MrProGetLength()) > 0) {
			if (NULL != (stData.m_puszData = (UCHAR*)MrMalloc(sizeof(UCHAR) * CPro.MrProGetLength()))) {
				if (stData.m_iLength == CTcp.MrNetRecvAll(stData.m_puszData, stData.m_iLength)) {
					memset(stKv, '\0', sizeof(MRPROKV_S) * MR_MR_VAL);
					MrProKvInput(stKv, CPro.MrProGetTotal(), stData.m_puszData, stData.m_iLength);
					while (usTotal < CPro.MrProGetTotal()) {
						switch (stKv[usTotal].m_ucKey) {
							case MR_MR_KEY:
								memcpy(szKey, stKv[usTotal].m_puszValue, stKv[usTotal].m_iLength);
								ullKey = strtoull(szKey, 0, 10);
								if (true == g_bDebug) {
									MR_DEBUG("Key[%d]: %llu\n", stKv[usTotal].m_iLength, ullKey);
								}
								break;
							case MR_MR_VAL:
								stData.m_ulStamp = time(NULL);
								if (true == g_bDebug) {
									MR_DEBUG("Val[%d]: ********************\n", stKv[usTotal].m_iLength);
									MR_DEBUG("Tim: %ld\n", stData.m_ulStamp);
								}
								break;
							default:
								break;
						}
						usTotal += 1;
					}

					if (0 == ullKey) {
						MR_ERROR("Key error\n");
						ucResult = MR_DEF_VLER;
					}
				}
				else {
					MR_ERROR("Failed to MrNetRecvAll\n");
					MrFree((void**)&stData.m_puszData);
					ucResult = MR_DEF_RECV;
				}

				if (MR_DEF_SELT == CPro.MrProGetVerb()) {
					MrFree((void**)&stData.m_puszData);
				}
			}
			else {
				MR_ERROR("Failed to MrMalloc\n");
				ucResult = MR_DEF_MALC;
			}
		}
		else {
			MR_ERROR("Failed to MrProGetLength\n");
			ucResult = MR_DEF_LENG;
		}

		/* hash */
		if (MR_DEF_SUCC == ucResult) {
			ppCTmap = (CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>**)g_CThash.MrHashGetBucket(ullKey);
			switch (CPro.MrProGetVerb()) {
				case MR_DEF_INST:
					if (true == g_bDebug) {
						MR_DEBUG("Insert -> CMrThash bucket: %lld, ", ullKey % MR_HB_THASH);
					}
					if (NULL == *ppCTmap) {
						g_CThash.MrThashLock();
						if (NULL == *ppCTmap) {
							*ppCTmap = new CMrTmap<ULONGLONG, MRHUBDATA_S, MRMAPDIGIT_O>;
							g_CThash.MrHashPutBucket(ullKey, (void**)ppCTmap);
							(*ppCTmap)->MrTmapCreateLock();
							if (true == g_bDebug) {
								MR_PRINT("Create CMrTmap\n");
							}
						}
						g_CThash.MrThashUnlock();
					}
					else {
						if (true == g_bDebug) {
							MR_PRINT("CMrTmap exist\n");
						}
					}
					if (true != ((*ppCTmap)->MrTmapInsert(ullKey, stData))) {
						MR_ERROR("Failed to MrTmapInsert\n");
						ucResult = MR_DEF_INST;
					}
					break;
				case MR_DEF_SELT:
					if (true == g_bDebug) {
						MR_DEBUG("Select -> CMrThash bucket: %lld, ", ullKey % MR_HB_THASH);
					}
					if (NULL != *ppCTmap) {
						if (true == ((*ppCTmap)->MrTmapSelect(ullKey, stData, sizeof(MRHUBDATA_S)))) {
							if (true == g_bDebug) {
								MR_PRINT("stData.Stamp: %ld, ", stData.m_ulStamp);
								MR_PRINT("stData.Length: %d\n", stData.m_iLength);
							}
							(*ppCTmap)->MrTmapDelete(ullKey);
						}
						else {
							if (true == g_bDebug) {
								MR_PRINT("Key NULL\n");
							}
							ucResult = MR_DEF_NULL;
						}
					}
					else {
						if (true == g_bDebug) {
							MR_PRINT("CMrTmap NULL\n");
						}
						ucResult = MR_DEF_NULL;
					}
					break;
				default:
					break;
			}
		}

		/* resp */
		if (true == g_bDebug) {
			MR_DEBUG("Response result: 0x%02x\n", ucResult);
		}
		CPro.MrProPutBody(NULL);
		CPro.MrProPutReal(MR_PRO_0);
		CPro.MrProPutResult(ucResult);
		if (MR_DEF_SELT == CPro.MrProGetVerb()) {
			if (MR_DEF_SUCC == ucResult) {
				CPro.MrProPutTotal(MR_MR_VAL);
				CPro.MrProPutLength(stData.m_iLength);
			}
			else {
				CPro.MrProPutTotal(MR_PRO_0);
				CPro.MrProPutLength(MR_PRO_0);
			}
		}
		else {
			CPro.MrProPutTotal(MR_PRO_0);
			CPro.MrProPutLength(MR_PRO_0);
		}
		CPro.MrProCreate(uszHead, MR_PRO_HEAD);
		CTcp.MrNetSend(uszHead, MR_PRO_HEAD);

		if (MR_DEF_SELT == CPro.MrProGetVerb()) {
			if (NULL != stData.m_puszData) {
				if (true == g_bDebug) {
					MR_DEBUG("Response data length: %d\n", stData.m_iLength);
				}
				CTcp.MrNetSendAll(stData.m_puszData, stData.m_iLength);
				MrFree((void**)&stData.m_puszData);
			}
		}

		if (true == g_bDebug) {
			MR_DEBUG("Thread[%lu] end Use time ", pthread_self());
			CTime.MrTimeOff();
			CTime.MrTimeDisplay();
		}
	}

	CTcp.MrNetClose();

	pthread_exit(NULL);
}
